package com.example.demo.BlockingQueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created with IntelliJ IDEA.
 * Description: 双锁实现阻塞队列
 * User: sfx
 * Date: 2023-05-10
 * Time: 15:28
 */
public class MyBlockingQueue<E> implements BlockingQueue<E>{

    private E[] array;//容器
    private int head;//头指针
    private int tail;//尾指针
    //实际有效元素个数
    private AtomicInteger size;
    //阻塞队列容量
    private int capacity;

    //当阻塞队列满了加的锁
    private ReentrantLock tailLock = new ReentrantLock();
    //当阻塞队列为空加的锁
    private ReentrantLock headLock = new ReentrantLock();

    //当阻塞队列满了进入的条件变量
    private Condition tailWaits = tailLock.newCondition();
    //当阻塞队列为空进入的条件变量
    private Condition headWaits = headLock.newCondition();

    public MyBlockingQueue(int capacity) {
        this.array = (E[]) new Object[capacity];
        this.capacity = capacity;
        this.size = new AtomicInteger();
    }

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;//加入元素前的个数
        try{
            while(isFull()) {
                tailWaits.await();
            }
            this.array[tail] = e;
            if(++tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            if(c + 1 < array.length) {
                tailWaits.signal();
            }
        }finally {
            tailLock.unlock();
        }
        //从空-->非空,剩下的poll线程级联唤醒
        if(c == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;//加入元素前的个数
        long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
        try{
            while(isFull()) {// offer2 offer3
                if(nanos <=0) {
                    return false;
                }
                nanos = tailWaits.awaitNanos(nanos);
            }
            this.array[tail] = e;
            if(++tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            if(c + 1 < array.length) {
                tailWaits.signal();
            }
        }finally {
            tailLock.unlock();
        }
        //从空-->非空,剩下的poll线程级联唤醒
        if(c == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        headLock.lockInterruptibly();
        E e;
        int c; //取走前的元素个数
        try{
            while(isEmpty()) {//poll1 poll2 poll3
                headWaits.await();
            }
            e = this.array[head];
            this.array[head] = null;
            if(++head == array.length) {
                head = 0;
            }
            c = size.getAndDecrement();
            if(c>1) {
                headWaits.signal();
            }
        }finally {
            headLock.unlock();
        }
        //从满-->不满
        if(c == array.length) {
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    //判断阻塞队列是否满
    private boolean isFull() {
        return this.size.get() == this.array.length;
    }
    //判断阻塞队列是否空
    private boolean isEmpty() {
        return this.size.get() == 0;
    }
    @Override
    public String toString() {
        return "MyBlockingQueue{" +
                "queue=" + Arrays.toString(array);
    }
}